/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements. See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.kafka.clients;

import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.AuthenticationException;
import org.apache.kafka.common.errors.UnsupportedVersionException;
import org.apache.kafka.common.metrics.Sensor;
import org.apache.kafka.common.network.ChannelState;
import org.apache.kafka.common.network.NetworkReceive;
import org.apache.kafka.common.network.Selectable;
import org.apache.kafka.common.network.Send;
import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.protocol.CommonFields;
import org.apache.kafka.common.protocol.types.Struct;
import org.apache.kafka.common.requests.AbstractRequest;
import org.apache.kafka.common.requests.AbstractResponse;
import org.apache.kafka.common.requests.ApiVersionsRequest;
import org.apache.kafka.common.requests.ApiVersionsResponse;
import org.apache.kafka.common.requests.MetadataRequest;
import org.apache.kafka.common.requests.MetadataResponse;
import org.apache.kafka.common.requests.RequestHeader;
import org.apache.kafka.common.requests.ResponseHeader;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.common.utils.Utils;
import org.slf4j.Logger;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.stream.Collectors;

/**
 * A network client for asynchronous request/response network i/o. This is an internal class used to implement the
 * user-facing producer and consumer clients.
 * <p>
 * This class is not thread-safe!
 */
public class NetworkClient implements KafkaClient {

    private final Logger log;

    /* 用于执行网络i/o的选择器 */
    private final Selectable selector;

    private final MetadataUpdater metadataUpdater;

    private final Random randOffset;

    /** 保存与每个节点的连接状态 */
    private final ClusterConnectionStates connectionStates;

    /** 请求队列，保存正在发送但还没有收到响应的请求 */
    private final InFlightRequests inFlightRequests;

    /* 套接字发送缓冲区字节大小 */
    private final int socketSendBuffer;

    /* 套接字接收缓冲区字节大小*/
    private final int socketReceiveBuffer;

    /* the client id used to identify this client in requests to the server */
    private final String clientId;

    /* the current correlation id to use when sending requests to servers */
    private int correlation;

    /* default timeout for individual requests to await acknowledgement from servers */
    private final int defaultRequestTimeoutMs;

    /* time in ms to wait before retrying to create connection to a server */
    private final long reconnectBackoffMs;

    private final Time time;

    /**
     * 是否需要与服务器的版本协调，默认都为true.第一次与broker连接需要
     */
    private final boolean discoverBrokerVersions;

    private final ApiVersions apiVersions;
    /**
     * 存储着要发送的版本请求，Key为主机地址，Value为构建请求的Builder
     */
    private final Map<String, ApiVersionsRequest.Builder> nodesNeedingApiVersionsFetch = new HashMap<>();

    private final List<ClientResponse> abortedSends = new LinkedList<>();

    private final Sensor throttleTimeSensor;

    public NetworkClient(Selectable selector,
                         Metadata metadata,
                         String clientId,
                         int maxInFlightRequestsPerConnection,
                         long reconnectBackoffMs,
                         long reconnectBackoffMax,
                         int socketSendBuffer,
                         int socketReceiveBuffer,
                         int defaultRequestTimeoutMs,
                         Time time,
                         boolean discoverBrokerVersions,
                         ApiVersions apiVersions,
                         LogContext logContext) {
        this(null,
             metadata,
             selector,
             clientId,
             maxInFlightRequestsPerConnection,
             reconnectBackoffMs,
             reconnectBackoffMax,
             socketSendBuffer,
             socketReceiveBuffer,
             defaultRequestTimeoutMs,
             time,
             discoverBrokerVersions,
             apiVersions,
             null,
             logContext);
    }

    public NetworkClient(Selectable selector,
            Metadata metadata,
            String clientId,
            int maxInFlightRequestsPerConnection,
            long reconnectBackoffMs,
            long reconnectBackoffMax,
            int socketSendBuffer,
            int socketReceiveBuffer,
            int defaultRequestTimeoutMs,
            Time time,
            boolean discoverBrokerVersions,
            ApiVersions apiVersions,
            Sensor throttleTimeSensor,
            LogContext logContext) {
        this(null,
             metadata,
             selector,
             clientId,
             maxInFlightRequestsPerConnection,
             reconnectBackoffMs,
             reconnectBackoffMax,
             socketSendBuffer,
             socketReceiveBuffer,
             defaultRequestTimeoutMs,
             time,
             discoverBrokerVersions,
             apiVersions,
             throttleTimeSensor,
             logContext);
    }

    public NetworkClient(Selectable selector,
                         MetadataUpdater metadataUpdater,
                         String clientId,
                         int maxInFlightRequestsPerConnection,
                         long reconnectBackoffMs,
                         long reconnectBackoffMax,
                         int socketSendBuffer,
                         int socketReceiveBuffer,
                         int defaultRequestTimeoutMs,
                         Time time,
                         boolean discoverBrokerVersions,
                         ApiVersions apiVersions,
                         LogContext logContext) {
        this(metadataUpdater,
             null,
             selector,
             clientId,
             maxInFlightRequestsPerConnection,
             reconnectBackoffMs,
             reconnectBackoffMax,
             socketSendBuffer,
             socketReceiveBuffer,
             defaultRequestTimeoutMs,
             time,
             discoverBrokerVersions,
             apiVersions,
             null,
             logContext);
    }

    private NetworkClient(MetadataUpdater metadataUpdater,
                          Metadata metadata,
                          Selectable selector,
                          String clientId,
                          int maxInFlightRequestsPerConnection,
                          long reconnectBackoffMs,
                          long reconnectBackoffMax,
                          int socketSendBuffer,
                          int socketReceiveBuffer,
                          int defaultRequestTimeoutMs,
                          Time time,
                          boolean discoverBrokerVersions,
                          ApiVersions apiVersions,
                          Sensor throttleTimeSensor,
                          LogContext logContext) {
        /* It would be better if we could pass `DefaultMetadataUpdater` from the public constructor, but it's not
         * possible because `DefaultMetadataUpdater` is an inner class and it can only be instantiated after the
         * super constructor is invoked.
         */
        if (metadataUpdater == null) {
            if (metadata == null) {
                throw new IllegalArgumentException("`metadata` must not be null");
            }
            this.metadataUpdater = new DefaultMetadataUpdater(metadata);
        } else {
            this.metadataUpdater = metadataUpdater;
        }
        this.selector = selector;
        this.clientId = clientId;
        this.inFlightRequests = new InFlightRequests(maxInFlightRequestsPerConnection);
        this.connectionStates = new ClusterConnectionStates(reconnectBackoffMs, reconnectBackoffMax);
        this.socketSendBuffer = socketSendBuffer;
        this.socketReceiveBuffer = socketReceiveBuffer;
        this.correlation = 0;
        this.randOffset = new Random();
        this.defaultRequestTimeoutMs = defaultRequestTimeoutMs;
        this.reconnectBackoffMs = reconnectBackoffMs;
        this.time = time;
        this.discoverBrokerVersions = discoverBrokerVersions;
        this.apiVersions = apiVersions;
        this.throttleTimeSensor = throttleTimeSensor;
        this.log = logContext.logger(NetworkClient.class);
    }

    /**
     * 开始连接到给定节点，如果已经连接并准备发送到该节点，则返回true。
     *
     * 一旦有读操作，就要读取一个完整的NetworkReceive。 如果是写，可以分多次写。
     * 即读操作会在一次SelectionKey循环读取一个完整的接收动作，而写操作会在多次SelectionKey中完成一个完整的发送动作。
     * 这里ready相关，一个channel里面的Send对象要是只发送了部分，下次就不会处于ready状态了。
     *
     * @param node The node to check
     * @param now The current timestamp
     * @return True if we are ready to send to the given node
     */
    @Override
    public boolean ready(Node node, long now) {
        // 检查的当前节点为空，报异常
        if (node.isEmpty()) {
            throw new IllegalArgumentException("Cannot connect to empty node " + node);
        }

        // 判断是否准备就绪，可以发送数据
        if (isReady(node, now)) {
            return true;
        }

        // 如果第一次运行，没有建立好，建立网路连接
        if (connectionStates.canConnect(node.idString(), now)) {
            initiateConnect(node, now); // 初始化连接
        }

        return false;
    }

    // Visible for testing
    boolean canConnect(Node node, long now) {
        return connectionStates.canConnect(node.idString(), now);
    }

    /**
     * Disconnects the connection to a particular node, if there is one.
     * Any pending ClientRequests for this connection will receive disconnections.
     *
     * @param nodeId The id of the node
     */
    @Override
    public void disconnect(String nodeId) {
        if (connectionStates.isDisconnected(nodeId))
            return;

        selector.close(nodeId);
        List<ApiKeys> requestTypes = new ArrayList<>();
        long now = time.milliseconds();
        for (InFlightRequest request : inFlightRequests.clearAll(nodeId)) {
            if (request.isInternalRequest) {
                if (request.header.apiKey() == ApiKeys.METADATA) {
                    metadataUpdater.handleDisconnection(request.destination);
                }
            } else {
                requestTypes.add(request.header.apiKey());
                abortedSends.add(new ClientResponse(request.header,
                        request.callback, request.destination, request.createdTimeMs, now,
                        true, null, null, null));
            }
        }
        connectionStates.disconnected(nodeId, now);
        if (log.isDebugEnabled()) {
            log.debug("Manually disconnected from {}. Removed requests: {}.", nodeId,
                Utils.join(requestTypes, ", "));
        }
    }

    /**
     * Closes the connection to a particular node (if there is one).
     * All requests on the connection will be cleared.  ClientRequest callbacks will not be invoked
     * for the cleared requests, nor will they be returned from poll().
     *
     * @param nodeId The id of the node
     */
    @Override
    public void close(String nodeId) {
        selector.close(nodeId);
        for (InFlightRequest request : inFlightRequests.clearAll(nodeId))
            if (request.isInternalRequest && request.header.apiKey() == ApiKeys.METADATA)
                metadataUpdater.handleDisconnection(request.destination);
        connectionStates.remove(nodeId);
    }

    /**
     * Returns the number of milliseconds to wait, based on the connection state, before attempting to send data. When
     * disconnected, this respects the reconnect backoff time. When connecting or connected, this handles slow/stalled
     * connections.
     *
     * @param node The node to check
     * @param now The current timestamp
     * @return The number of milliseconds to wait.
     */
    @Override
    public long connectionDelay(Node node, long now) {
        return connectionStates.connectionDelay(node.idString(), now);
    }

    // Return the remaining throttling delay in milliseconds if throttling is in progress. Return 0, otherwise.
    // This is for testing.
    public long throttleDelayMs(Node node, long now) {
        return connectionStates.throttleDelayMs(node.idString(), now);
    }

    /**
     * Return the poll delay in milliseconds based on both connection and throttle delay.
     * @param node the connection to check
     * @param now the current time in ms
     */
    @Override
    public long pollDelayMs(Node node, long now) {
        return connectionStates.pollDelayMs(node.idString(), now);
    }

    /**
     * Check if the connection of the node has failed, based on the connection state. Such connection failure are
     * usually transient and can be resumed in the next {@link #ready(org.apache.kafka.common.Node, long)} }
     * call, but there are cases where transient failures needs to be caught and re-acted upon.
     *
     * @param node the node to check
     * @return true iff the connection has failed and the node is disconnected
     */
    @Override
    public boolean connectionFailed(Node node) {
        return connectionStates.isDisconnected(node.idString());
    }

    /**
     * Check if authentication to this node has failed, based on the connection state. Authentication failures are
     * propagated without any retries.
     *
     * @param node the node to check
     * @return an AuthenticationException iff authentication has failed, null otherwise
     */
    @Override
    public AuthenticationException authenticationException(Node node) {
        return connectionStates.authenticationException(node.idString());
    }

    /**
     * 检查具有给定id的节点是否准备好发送更多请求。
     *
     * @param node The node
     * @param now The current time in ms
     * @return true if the node is ready
     */
    @Override
    public boolean isReady(Node node, long now) {
        // !metadataUpdater.isUpdateDue(now): 判断metadata是否到了更新的时候了，如果metadata需要更新，那么就不发送本次请求，也就是metadata更新优先级高
        // canSendRequest(node.idString(), now)：能够发送请求(条件是connection就绪，对应的channel就绪,inFlightRequests小于配置的最大值)
        return !metadataUpdater.isUpdateDue(now) && canSendRequest(node.idString(), now);
    }

    /**
     * 我们是否已连接并准备好向给定连接发送更多请求？
     * @param node The node
     * @param now the current timestamp
     */
    private boolean canSendRequest(String node, long now) {
        // connectionStates.isReady(node, now)：连接准备好
        // selector.isChannelReady(node): node节点的channel是否建立好
        // inFlightRequests.canSendMore(node): 检测发送到broker的消息，还没有收到响应的个数是否超过FlightRequest的最大值(默认5个)
        return connectionStates.isReady(node, now) &&
                selector.isChannelReady(node) &&
                inFlightRequests.canSendMore(node);
    }

    /**
     * Queue up the given request for sending. Requests can only be sent out to ready nodes.
     * @param request The request
     * @param now The current timestamp
     */
    @Override
    public void send(ClientRequest request, long now) {
        doSend(request, false, now);
    }

    private void sendInternalMetadataRequest(MetadataRequest.Builder builder, String nodeConnectionId, long now) {
        // 将MetadataRequest包装成clientRequest
        ClientRequest clientRequest = newClientRequest(nodeConnectionId, builder, now, true);
        // 最终调用selector的send
        doSend(clientRequest, true, now);
    }

    /**
     * Producer 端的请求都是通过 NetworkClient.dosend() 来发送的，其作用就是：
     * 1. 检查版本信息，并根据 apiKey() 构建 Request；
     * 2. 创建 NetworkSend 实例；
     * 3. 调用 Selector.send 发送该 Send。
     *
     * @param clientRequest
     * @param isInternalRequest
     * @param now
     */
    private void doSend(ClientRequest clientRequest, boolean isInternalRequest, long now) {
        String nodeId = clientRequest.destination();
        if (!isInternalRequest) {
            // 如果此请求来自NetworkClient外部，请验证我们是否可以发送数据。
            // 如果请求是内部的，我们相信内部代码已经完成了验证。 对于某些内部请求，验证将略有不同 (例如, ApiVersionsRequests可以在处于 READY 状态之前发送)
            // 检测是否能够向指定Node发送请求
            if (!canSendRequest(nodeId, now)) {
                throw new IllegalStateException("Attempt to send a request to node " + nodeId + " which is not ready.");
            }
        }
        AbstractRequest.Builder<?> builder = clientRequest.requestBuilder();
        // 构建 AbstractRequest, 检查其版本信息
        try {
            // 检测版本
            NodeApiVersions versionInfo = apiVersions.get(nodeId);
            short version;
            // Note: 如果versionInfo为空，则没有服务器版本信息.
            // 当发送初始ApiVersionRequest（获取版本信息本身）时会出现这种情况。
            // discoverbrokervisions设置为false时也是如此。
            if (versionInfo == null) {
                version = builder.latestAllowedVersion();
                if (discoverBrokerVersions && log.isTraceEnabled()) {
                    log.trace("No version information found when sending {} with correlation id {} to node {}. " + "Assuming version {}.", clientRequest.apiKey(), clientRequest.correlationId(), nodeId, version);
                }
            } else {
                version = versionInfo.latestUsableVersion(clientRequest.apiKey(), builder.oldestAllowedVersion(), builder.latestAllowedVersion());
            }
            // 如果所选版本中存在无法表示的基本字段. 对build的调用也可能抛出UnsupportedVersionException,
            // builder.build()是ProduceRequest.Builder ，结果是ProduceRequest
            doSend(clientRequest, isInternalRequest, now, builder.build(version));
        } catch (UnsupportedVersionException unsupportedVersionException) {
            // 版本不支持，不发送，生成clientResponse，添加到abortedSends集合里
            log.debug("Version mismatch when attempting to send {} with correlation id {} to {}", builder, clientRequest.correlationId(), clientRequest.destination(), unsupportedVersionException);
            ClientResponse clientResponse = new ClientResponse(clientRequest.makeHeader(builder.latestAllowedVersion()),
                    clientRequest.callback(), clientRequest.destination(), now, now, false, unsupportedVersionException, null, null);
            abortedSends.add(clientResponse);
        }
    }

    private void doSend(ClientRequest clientRequest, boolean isInternalRequest, long now, AbstractRequest request) {
        // 目标节点
        String destination = clientRequest.destination();
        // 生成请求头
        RequestHeader header = clientRequest.makeHeader(request.version());
        if (log.isDebugEnabled()) {
            int latestClientVersion = clientRequest.apiKey().latestVersion();
            if (header.apiVersion() == latestClientVersion) {
                log.trace("Sending {} {} with correlation id {} to node {}", clientRequest.apiKey(), request, clientRequest.correlationId(), destination);
            } else {
                log.debug("Using older server API v{} to send {} {} with correlation id {} to node {}", header.apiVersion(), clientRequest.apiKey(), request, clientRequest.correlationId(), destination);
            }
        }

        // toSend()是一个接口，返回的是NetworkSend（继承了ByteBufferSend
        // 结合请求头和请求体，序列化数据，保存到NetworkSend
        Send send = request.toSend(destination, header);
        // 生成InFlightRequest实例，它保存了正在发送中的请求
        InFlightRequest inFlightRequest = new InFlightRequest(
                clientRequest,
                header,
                isInternalRequest,
                request,
                send,
                now);

        // 先加入到inFlightRequests队列中
        this.inFlightRequests.add(inFlightRequest);
        // 调用Selector异步发送数据（将send和对应kafkaChannel绑定起来，并开启该kafkaChannel底层socket的写事件）
        selector.send(send);
    }

    /**
     * Do actual reads and writes to sockets.
     *
     * @param timeout The maximum amount of time to wait (in ms) for responses if there are none immediately,
     *                must be non-negative. The actual timeout will be the minimum of timeout, request timeout and
     *                metadata timeout
     * @param now The current time in milliseconds
     * @return The list of responses received
     */
    @Override
    public List<ClientResponse> poll(long timeout, long now) {
        if (!abortedSends.isEmpty()) {
            // 如果由于不支持的版本异常或断开连接而中止发送，需要立即处理
            List<ClientResponse> responses = new ArrayList<>();
            handleAbortedSends(responses);
            completeResponses(responses);
            return responses;
        }

        // 看是否需要先更新集群的元数据 ， metadataUpdater，专门更新元数据的组件
        long metadataTimeout = metadataUpdater.maybeUpdate(now);
        try {
            // 调用selector的poll方法， 真正读写发生的地方. 如果客户端请求被完整地处理过了, 会加入到completeSends或complteReceives中
            this.selector.poll(Utils.min(timeout, metadataTimeout, defaultRequestTimeoutMs));
        } catch (IOException e) {
            log.error("Unexpected error during I/O", e);
        }

        // 处理完成的动作
        long updatedNow = this.time.milliseconds();
        List<ClientResponse> responses = new ArrayList<>();
        handleCompletedSends(responses, updatedNow);    // 处理完成的发送响应
        handleCompletedReceives(responses, updatedNow); // 处理完成的接收响应
        handleDisconnections(responses, updatedNow);    // 处理关闭的响应
        handleConnections();                            // 处理连接状态变化
        handleInitiateApiVersionRequests(updatedNow);   // 处理初始化ApiVersion请求
        handleTimedOutRequests(responses, updatedNow);
        // 将已完成的请求，生成ClientReponse收集起来,
        completeResponses(responses);

        return responses;
    }

    private void completeResponses(List<ClientResponse> responses) {
        for (ClientResponse response : responses) {
            try {
                // 调用response的完成
                response.onComplete();
            } catch (Exception e) {
                log.error("Uncaught error in request completion:", e);
            }
        }
    }

    /**
     * Get the number of in-flight requests
     */
    @Override
    public int inFlightRequestCount() {
        return this.inFlightRequests.count();
    }

    @Override
    public boolean hasInFlightRequests() {
        return !this.inFlightRequests.isEmpty();
    }

    /**
     * Get the number of in-flight requests for a given node
     */
    @Override
    public int inFlightRequestCount(String node) {
        return this.inFlightRequests.count(node);
    }

    @Override
    public boolean hasInFlightRequests(String node) {
        return !this.inFlightRequests.isEmpty(node);
    }

    @Override
    public boolean hasReadyNodes(long now) {
        return connectionStates.hasReadyNodes(now);
    }

    /**
     * Interrupt the client if it is blocked waiting on I/O.
     */
    @Override
    public void wakeup() {
        this.selector.wakeup();
    }

    /**
     * Close the network client
     */
    @Override
    public void close() {
        this.selector.close();
        this.metadataUpdater.close();
    }

    /**
     * 选择未完成请求最少的节点，该节点至少符合连接条件。
     * 此方法将选择具有现有连接的节点，但如果所有现有连接都在使用，则可能选择尚未具有连接的节点。
     * 此方法永远不会选择没有现有连接且在重新连接回退期间已断开连接的节点。
     *
     * @return The node with the fewest in-flight requests.
     */
    @Override
    public Node leastLoadedNode(long now) {
        // 获取当前集群所有节点
        List<Node> nodes = this.metadataUpdater.fetchNodes();
        int inflight = Integer.MAX_VALUE;
        Node found = null;

        // 集合大小内随机取一个整数，作为偏移量
        int offset = this.randOffset.nextInt(nodes.size());
        for (int i = 0; i < nodes.size(); i++) {
            // 计算节点id
            int idx = (offset + i) % nodes.size();
            // 获取该节点
            Node node = nodes.get(idx);
            // 获取该节点上现有的请求数
            int currInflight = this.inFlightRequests.count(node.idString());
            // 如果没有请求数 并且 已经建立好连接等待的话
            if (currInflight == 0 && isReady(node, now)) {
                log.trace("Found least loaded node {} connected with no in-flight requests", node);
                // 返回此节点
                return node;
            }
            // 如果节点是可连接的 并且 请求数小于目前已知节点最小的请求数的话
            else if (!this.connectionStates.isBlackedOut(node.idString(), now) && currInflight < inflight) {
                // 更新最小请求数
                inflight = currInflight;
                // 记录此节点
                found = node;
            }
            // 否则，打印日志，此节点不是我们要寻找的节点
            else if (log.isTraceEnabled()) {
                log.trace("Removing node {} from least loaded node selection: is-blacked-out: {}, in-flight-requests: {}", node, this.connectionStates.isBlackedOut(node.idString(), now), currInflight);
            }
        }

        if (found != null) {
            log.trace("Found least loaded node {}", found);
        } else {
            log.trace("Least loaded node selection failed to find an available node");
        }
        // 返回记录的请求数最少的节点
        return found;
    }

    public static AbstractResponse parseResponse(ByteBuffer responseBuffer, RequestHeader requestHeader) {
        Struct responseStruct = parseStructMaybeUpdateThrottleTimeMetrics(responseBuffer, requestHeader, null, 0);
        return AbstractResponse.parseResponse(requestHeader.apiKey(), responseStruct);
    }

    private static Struct parseStructMaybeUpdateThrottleTimeMetrics(ByteBuffer responseBuffer, RequestHeader requestHeader,
                                                                    Sensor throttleTimeSensor, long now) {
        // 解析响应头
        ResponseHeader responseHeader = ResponseHeader.parse(responseBuffer);
        // 解析响应体，总是期望响应版本id与请求版本id相同
        Struct responseBody = requestHeader.apiKey().parseResponse(requestHeader.apiVersion(), responseBuffer);
        // 验证请求头与响应头的 correlation id 必须相等
        correlate(requestHeader, responseHeader);
        if (throttleTimeSensor != null && responseBody.hasField(CommonFields.THROTTLE_TIME_MS))
            throttleTimeSensor.record(responseBody.get(CommonFields.THROTTLE_TIME_MS), now);
        return responseBody;
    }

    /**
     * Post process disconnection of a node
     *
     * @param responses The list of responses to update
     * @param nodeId Id of the node to be disconnected
     * @param now The current time
     */
    private void processDisconnection(List<ClientResponse> responses,
                                      String nodeId,
                                      long now,
                                      ChannelState disconnectState) {
        // 修改连接状态
        connectionStates.disconnected(nodeId, now);
        apiVersions.remove(nodeId);
        nodesNeedingApiVersionsFetch.remove(nodeId);
        switch (disconnectState.state()) {
            case AUTHENTICATION_FAILED:
                AuthenticationException exception = disconnectState.exception();
                connectionStates.authenticationFailed(nodeId, now, exception);
                metadataUpdater.handleAuthenticationFailure(exception);
                log.error("Connection to node {} failed authentication due to: {}", nodeId, exception.getMessage());
                break;
            case AUTHENTICATE:
                log.warn("Connection to node {} terminated during authentication. This may indicate " + "that authentication failed due to invalid credentials.", nodeId);
                break;
            case NOT_CONNECTED:
                log.warn("Connection to node {} could not be established. Broker may not be available.", nodeId);
                break;
            default:
                break;
        }

        for (InFlightRequest request : this.inFlightRequests.clearAll(nodeId)) {
            log.trace("Cancelled request {} {} with correlation id {} due to node {} being disconnected", request.header.apiKey(), request.request, request.header.correlationId(), nodeId);
            if (!request.isInternalRequest) {
                responses.add(request.disconnected(now, disconnectState.exception()));
            } else if (request.header.apiKey() == ApiKeys.METADATA) {
                metadataUpdater.handleDisconnection(request.destination);
            }
        }
    }

    /**
     * Iterate over all the inflight requests and expire any requests that have exceeded the configured requestTimeout.
     * The connection to the node associated with the request will be terminated and will be treated as a disconnection.
     *
     * @param responses The list of responses to update
     * @param now The current time
     */
    private void handleTimedOutRequests(List<ClientResponse> responses, long now) {
        // 获取到请求超时的主机。
        List<String> nodeIds = this.inFlightRequests.nodesWithTimedOutRequests(now);
        for (String nodeId : nodeIds) {
            // 关闭请求超时的主机的连接
            this.selector.close(nodeId);
            log.debug("Disconnecting from node {} due to request timeout.", nodeId);
            //我们猜应该是会去修改 连接的状态
            processDisconnection(responses, nodeId, now, ChannelState.LOCAL_CLOSE);
        }

        if (!nodeIds.isEmpty()) {
            metadataUpdater.requestUpdate();
        }
    }

    private void handleAbortedSends(List<ClientResponse> responses) {
        responses.addAll(abortedSends);
        abortedSends.clear();
    }

    /**
     * 客户端发送请求后，handleCompletedSends中对于有响应的请求，并不会将ClientRequest从inFlightRequests中移除。（因为没收到响应）
     * 除非是客户端请求不需要响应，则这时候是可以将ClientRequest从中删除，添加时放到头部，删除时也是从头部删除。
     *
     * 怎么能保证从Selector返回的请求，是对应到队列中最新的请求?
     * InFlightRequests 中存储的是已发送但没收到响应的请求，
     * 而 completedSends 保存的是最近一次 poll() 方法中发送成功的请求（发送成功并没有收到响应）.
     * 每个请求发送，都要等待前面的请求发送完成，这样就能保证同一时间只有一个请求正在发送 。这样就能对应上。
     * https://img-blog.csdnimg.cn/20190330231053986.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JvaHU4Mw==
     *
     * @param responses The list of responses to update
     * @param now The current time
     */
    private void handleCompletedSends(List<ClientResponse> responses, long now) {
        // if no response is expected then when the send is completed, return it
        for (Send send : this.selector.completedSends()) {
            // 实际上就是在刚刚的那个poll()方法里，对一个broker发送出去的request。
            // 这里获取目标节点的队列中第一个请求, 但并没有从队列中删除, 取出之后判断这个请求是否期望得到响应
            InFlightRequest request = this.inFlightRequests.lastSent(send.destination());
            // 如果不需要响应,当Send请求完成时,就直接返回.还是有request.completed生成的ClientResponse对象
            if (!request.expectResponse) {
                this.inFlightRequests.completeLastSent(send.destination());
                // 调用completed方法生成ClientResponse,completed（）第一个参数为null,表示没有响应内容
                responses.add(request.completed(null, now));
            }
            // 如果客户端请求需要有响应, 那么它的响应是在下面的handleCompletedReceives中设置的.
        }
    }

    /**
     * If a response from a node includes a non-zero throttle delay and client-side throttling has been enabled for
     * the connection to the node, throttle the connection for the specified delay.
     *
     * @param response the response
     * @param apiVersion the API version of the response
     * @param nodeId the id of the node
     * @param now The current time
     */
    private void maybeThrottle(AbstractResponse response, short apiVersion, String nodeId, long now) {
        int throttleTimeMs = response.throttleTimeMs();
        if (throttleTimeMs > 0 && response.shouldClientThrottle(apiVersion)) {
            connectionStates.throttle(nodeId, now + throttleTimeMs);
            log.trace("Connection to node {} is throttled for {} ms until timestamp {}", nodeId, throttleTimeMs,
                      now + throttleTimeMs);
        }
    }

    /**
     * Handle any completed receives and update the response list with the responses received.
     *
     * 收到响应是在handleCompletedReceives，这时候才可以调用completeNext删除source对应的ClientRequest，
     * 因为我们知道inFlightRequests存的是未收到请求的ClientRequest，现在这个请求已经有响应了，就不需要再其中保存了。
     * 那么inFlightRequests就起到了可以防止请求堆积的作用
     *
     * @param responses The list of responses to update
     * @param now The current time
     */
    private void handleCompletedReceives(List<ClientResponse> responses, long now) {
        // 遍历响应，通过Selector返回未处理的响应
        for (NetworkReceive receive : this.selector.completedReceives()) {
            String source = receive.source();
            // 从缓存队列获取已发送请求并删除
            // （这里会从inFlightRequests删除，因为inFlightRequests存的是未收到请求的ClientRequest，现在这个请求已经有响应了，就不需要再其中保存了。）
            // source对应的节点的inFlightRequestCount 原子减一
            InFlightRequest req = inFlightRequests.completeNext(source);
            // 解析响应,并且验证响应头，生成Struct实例
            Struct responseStruct = parseStructMaybeUpdateThrottleTimeMetrics(receive.payload(), req.header, throttleTimeSensor, now);

            if (log.isTraceEnabled()) {
                log.trace("Completed receive from node {} for {} with correlation id {}, received {}", req.destination,
                    req.header.apiKey(), req.header.correlationId(), responseStruct);
            }
            // 如果接收到的响应包括限制延迟，则限制连接。
            // 生成响应体
            AbstractResponse body = AbstractResponse.parseResponse(req.header.apiKey(), responseStruct);
            maybeThrottle(body, req.header.apiVersion(), req.destination, now);
            // 判断是否为MetadataResponse
            if (req.isInternalRequest && body instanceof MetadataResponse) {
                // 处理元数据请求的响应。
                metadataUpdater.handleCompletedMetadataResponse(req.header, now, (MetadataResponse) body);
            } else if (req.isInternalRequest && body instanceof ApiVersionsResponse) {
                // 处理办理请求响应ApiVersionsResponse
                handleApiVersionsResponse(responses, req, now, (ApiVersionsResponse) body);
            } else {
                // 其他响应：生成ClientResponse添加到列表中
                responses.add(req.completed(body, now));
            }
        }
    }

    /**
     * 处理版本请求响应
     * @param responses
     * @param req
     * @param now
     * @param apiVersionsResponse
     */
    private void handleApiVersionsResponse(List<ClientResponse> responses,
                                           InFlightRequest req, long now, ApiVersionsResponse apiVersionsResponse) {
        // 目标节点
        final String node = req.destination;
        // 判断响应和版本是否协调
        if (apiVersionsResponse.error() != Errors.NONE) {
            // 异常处理
            if (req.request.version() == 0 || apiVersionsResponse.error() != Errors.UNSUPPORTED_VERSION) {
                log.warn("Received error {} from node {} when making an ApiVersionsRequest with correlation id {}. Disconnecting.",
                        apiVersionsResponse.error(), node, req.header.correlationId());
                this.selector.close(node);
                processDisconnection(responses, node, now, ChannelState.LOCAL_CLOSE);
            } else {
                nodesNeedingApiVersionsFetch.put(node, new ApiVersionsRequest.Builder((short) 0));
            }
            return;
        }
        NodeApiVersions nodeVersionInfo = new NodeApiVersions(apiVersionsResponse.apiVersions());
        // 更新版本
        apiVersions.update(node, nodeVersionInfo);
        // 连接状态更新为ready，可以正常发送请求了。
        this.connectionStates.ready(node);
        log.debug("Recorded API versions for node {}: {}", node, nodeVersionInfo);
    }

    /**
     * Handle any disconnected connections
     *
     * @param responses The list of responses that completed with the disconnection
     * @param now The current time
     */
    private void handleDisconnections(List<ClientResponse> responses, long now) {
        for (Map.Entry<String, ChannelState> entry : this.selector.disconnected().entrySet()) {
            String node = entry.getKey();
            log.debug("Node {} disconnected.", node);
            processDisconnection(responses, node, now, entry.getValue());
        }
        // we got a disconnect so we should probably refresh our metadata and see if that broker is dead
        if (this.selector.disconnected().size() > 0){
            // 如果连接断开，重新拉取元数据
            metadataUpdater.requestUpdate();
        }
    }

    /**
     * Record any newly completed connections
     */
    private void handleConnections() {
        // 遍历刚创建完成的连接
        for (String node : this.selector.connected()) {
            // 我们现在连接成功了.  我们可能仍然无法发送请求的节点. 例如,
            // 如果启用了SSL，那么SSL握手将在建立连接后发生。
            // 因此，在尝试通过此连接发送之前，仍然需要检查isChannelReady。
            if (discoverBrokerVersions) {
                // 更新连接的状态为版本协调状态
                this.connectionStates.checkingApiVersions(node);
                // 将请求保存到nodesNeedingApiVersionsFetch集合里
                nodesNeedingApiVersionsFetch.put(node, new ApiVersionsRequest.Builder());
                log.debug("Completed connection to node {}. Fetching API versions.", node);
            } else {
                this.connectionStates.ready(node);
                log.debug("Completed connection to node {}. Ready.", node);
            }
        }
    }

    /**
     * poll()方法调用完 handleConnections() 之后，会调用handleInitiateApiVersionRequests()发送发送版本协调请求
     * 在handleCompletedReceives()方法里面会调用 handleApiVersionsResponse() 负责处理响应
     * @param now
     */
    private void handleInitiateApiVersionRequests(long now) {
        // 遍历请求集合nodesNeedingApiVersionsFetch(在handleConnections加入的)
        Iterator<Map.Entry<String, ApiVersionsRequest.Builder>> iter = nodesNeedingApiVersionsFetch.entrySet().iterator();
        while (iter.hasNext()) {
            Map.Entry<String, ApiVersionsRequest.Builder> entry = iter.next();
            String node = entry.getKey();
            // 判断是否允许发送请求
            if (selector.isChannelReady(node) && inFlightRequests.canSendMore(node)) {
                log.debug("Initiating API versions fetch from node {}.", node);
                ApiVersionsRequest.Builder apiVersionRequestBuilder = entry.getValue();
                // 调用newClientRequest生成请求
                ClientRequest clientRequest = newClientRequest(node, apiVersionRequestBuilder, now, true);
                // 发送请求
                doSend(clientRequest, true, now);
                iter.remove();
            }
        }
    }

    /**
     * Validate that the response corresponds to the request we expect or else explode
     */
    private static void correlate(RequestHeader requestHeader, ResponseHeader responseHeader) {
        if (requestHeader.correlationId() != responseHeader.correlationId())
            throw new IllegalStateException("Correlation id for response (" + responseHeader.correlationId()
                    + ") does not match request (" + requestHeader.correlationId() + "), request header: " + requestHeader);
    }

    /**
     * 创建连接到指定节点
     */
    private void initiateConnect(Node node, long now) {
        String nodeConnectionId = node.idString();
        try {
            log.debug("Initiating connection to node {}", node);
            // 更新连接状态为正在连接
            this.connectionStates.connecting(nodeConnectionId, now);
            // 调用selector异步连接
            selector.connect(nodeConnectionId,
                             new InetSocketAddress(node.host(), node.port()),
                             this.socketSendBuffer, // socket发送缓冲区字节大小
                             this.socketReceiveBuffer); // socket接收缓冲区字节大小
        } catch (IOException e) {
            /* attempt failed, we'll try again after the backoff */
            connectionStates.disconnected(nodeConnectionId, now);
            /* maybe the problem is our metadata, update it */
            metadataUpdater.requestUpdate();
            log.warn("Error connecting to node {}", node, e);
        }
    }

    class DefaultMetadataUpdater implements MetadataUpdater {

        /* the current cluster metadata */
        private final Metadata metadata;

        /* true iff there is a metadata request that has been sent and for which we have not yet received a response */
        private boolean metadataFetchInProgress;

        DefaultMetadataUpdater(Metadata metadata) {
            this.metadata = metadata;
            this.metadataFetchInProgress = false;
        }

        @Override
        public List<Node> fetchNodes() {
            return metadata.fetch().nodes();
        }

        @Override
        public boolean isUpdateDue(long now) {
            // 当前不能处于元数据加载的过程，而且下一次要更新元数据的间隔时间为0
            // 现在没有加载元数据，但是马上就应该要加载元数据了，如果对上述条件判断是非的话，要不然是正在加载元数据，或者是还没到加载元数据的时候
            return !this.metadataFetchInProgress && this.metadata.timeToNextUpdate(now) == 0;
        }

        @Override
        public long maybeUpdate(long now) {
            // should we update our metadata?
            // 下一次更新集群信息的时间是当前信息过期和可以更新当前信息的时间的最大值(即后退时间已经过去); 如果已请求更新，则过期时间为现在
            long timeToNextMetadataUpdate = metadata.timeToNextUpdate(now);
            // 如果有一个元数据请求已经发送，但我们还没有收到响应，则返回用于等待服务器确认的单个请求的默认超时
            long waitForMetadataFetch = this.metadataFetchInProgress ? defaultRequestTimeoutMs : 0;
            // 取两者较大值
            long metadataTimeout = Math.max(timeToNextMetadataUpdate, waitForMetadataFetch);

            if (metadataTimeout > 0) {
                return metadataTimeout;
            }

            // 请注意 这个方法 和 poll()的超时计算 都高度依赖 leastLoadedNode.
            // 选择未完成请求最少的节点
            Node node = leastLoadedNode(now);
            if (node == null) {
                // 由于没有可用的节点，因此放弃发送元数据请求
                log.debug("Give up sending metadata request since no node is available");
                // 返回重试创建与服务器的连接之前等待的时间（毫秒）
                return reconnectBackoffMs;
            }

            return maybeUpdate(now, node);
        }

        @Override
        public void handleDisconnection(String destination) {
            Cluster cluster = metadata.fetch();
            // 'processDisconnection' generates warnings for misconfigured bootstrap server configuration
            // resulting in 'Connection Refused' and misconfigured security resulting in authentication failures.
            // The warning below handles the case where connection to a broker was established, but was disconnected
            // before metadata could be obtained.
            if (cluster.isBootstrapConfigured()) {
                int nodeId = Integer.parseInt(destination);
                Node node = cluster.nodeById(nodeId);
                if (node != null)
                    log.warn("Bootstrap broker {} disconnected", node);
            }

            metadataFetchInProgress = false;
        }

        @Override
        public void handleAuthenticationFailure(AuthenticationException exception) {
            metadataFetchInProgress = false;
            if (metadata.updateRequested())
                metadata.failedUpdate(time.milliseconds(), exception);
        }

        @Override
        public void handleCompletedMetadataResponse(RequestHeader requestHeader, long now, MetadataResponse response) {
            this.metadataFetchInProgress = false;
            // 获取响应中的集群对象信息
            Cluster cluster = response.cluster();

            // 如果任何分区的leader缺少侦听器，请最多记录其中的10个分区以诊断代理配置问题。
            // 如果将侦听器动态添加到代理中，这可能是一个暂时的问题。
            List<TopicPartition> missingListenerPartitions = response.topicMetadata().stream().flatMap(topicMetadata ->
                topicMetadata.partitionMetadata().stream()
                    .filter(partitionMetadata -> partitionMetadata.error() == Errors.LISTENER_NOT_FOUND)
                    .map(partitionMetadata -> new TopicPartition(topicMetadata.topic(), partitionMetadata.partition())))
                .collect(Collectors.toList());

            if (!missingListenerPartitions.isEmpty()) {
                int count = missingListenerPartitions.size();
                log.warn("{} partitions have leader brokers without a matching listener, including {}",
                        count, missingListenerPartitions.subList(0, Math.min(10, count)));
            }

            // 错误响应码处理
            Map<String, Errors> errors = response.errors();
            if (!errors.isEmpty())
                log.warn("Error while fetching metadata with correlation id {} : {}", requestHeader.correlationId(), errors);

            // don't update the cluster if there are no valid nodes...the topic we want may still be in the process of being
            // created which means we will get errors and no nodes until it exists
            // 启动metadata的更新
            if (cluster.nodes().size() > 0) {
                this.metadata.update(cluster, response.unavailableTopics(), now);
            } else {
                log.trace("Ignoring empty metadata response with correlation id {}.", requestHeader.correlationId());
                this.metadata.failedUpdate(now, null);
            }
        }

        @Override
        public void requestUpdate() {
            this.metadata.requestUpdate();
        }

        @Override
        public void close() {
            this.metadata.close();
        }

        /**
         * Return true if there's at least one connection establishment is currently underway
         */
        private boolean isAnyNodeConnecting() {
            for (Node node : fetchNodes()) {
                if (connectionStates.isConnecting(node.idString())) {
                    return true;
                }
            }
            return false;
        }

        /**
         * 如果可以的话，在发送列表中添加一个元数据请求
         */
        private long maybeUpdate(long now, Node node) {
            // 获取节点id
            String nodeConnectionId = node.idString();

            // 我们是否已经连接并准备好向给定的连接发送更多的请求?
            if (canSendRequest(nodeConnectionId, now)) {
                // 将正在处理元数据发送请求的标识置为true
                this.metadataFetchInProgress = true;
                MetadataRequest.Builder metadataRequest;
                // 获取是否需要所有主题的元数据
                if (metadata.needMetadataForAllTopics()) {
                    // 获取所有主题的元数据请求
                    metadataRequest = MetadataRequest.Builder.allTopics();
                } else {
                    // 获取指定主题的元数据请求
                    metadataRequest = new MetadataRequest.Builder(new ArrayList<>(metadata.topics()), metadata.allowAutoTopicCreation());
                }

                log.debug("Sending metadata request {} to node {}", metadataRequest, node);
                // 向节点发送元数据请求
                sendInternalMetadataRequest(metadataRequest, nodeConnectionId, now);
                // 返回单个请求等待服务器确认的默认超时时间
                return defaultRequestTimeoutMs;
            }

            // 判断当前是否至少有一个连接建立
            if (isAnyNodeConnecting()) {
                // 返回重试创建与服务器的连接之前等待的时间（毫秒）
                return reconnectBackoffMs;
            }

            // 如果我们现在可以在该节点上启动一个连接（之前在该节点无连接），那么返回true
            if (connectionStates.canConnect(nodeConnectionId, now)) {
                log.debug("Initialize connection to node {} for sending metadata request", node);
                // 启动该节点的连接，比如打开通道，注册到选择器等
                initiateConnect(node, now);
                // 返回重试创建与服务器的连接之前等待的时间（毫秒）
                return reconnectBackoffMs;
            }

            // 已连接，但无法发送更多或正在连接
            // 无论哪种情况, 我们只需要等待一个网络事件，让我们知道所选的连接可能再次可用。
            return Long.MAX_VALUE;
        }

    }

    @Override
    public ClientRequest newClientRequest(String nodeId,
                                          AbstractRequest.Builder<?> requestBuilder,
                                          long createdTimeMs,
                                          boolean expectResponse) {
        return newClientRequest(nodeId, requestBuilder, createdTimeMs, expectResponse, defaultRequestTimeoutMs, null);
    }

    @Override
    public ClientRequest newClientRequest(String nodeId,
                                          AbstractRequest.Builder<?> requestBuilder,
                                          long createdTimeMs,
                                          boolean expectResponse,
                                          int requestTimeoutMs,
                                          RequestCompletionHandler callback) {
        return new ClientRequest(nodeId, requestBuilder, correlation++, clientId, createdTimeMs, expectResponse,
                defaultRequestTimeoutMs, callback);
    }

    public boolean discoverBrokerVersions() {
        return discoverBrokerVersions;
    }

    static class InFlightRequest {
        /**
         * 请求头
         */
        final RequestHeader header;
        /**
         * 表示这个请求要发送到哪个Broker节点上
         */
        final String destination;
        /**
         * 回调函数
         */
        final RequestCompletionHandler callback;
        /**
         * 是否需要服务端返回响应
         */
        final boolean expectResponse;
        /**
         * 请求体
         */
        final AbstractRequest request;
        final boolean isInternalRequest; // 用于标记NetworkClient在内部发起的请求
        final Send send;
        final long sendTimeMs;
        final long createdTimeMs;
        final long requestTimeoutMs;

        public InFlightRequest(ClientRequest clientRequest,
                               RequestHeader header,
                               boolean isInternalRequest,
                               AbstractRequest request,
                               Send send,
                               long sendTimeMs) {
            this(header,
                 clientRequest.requestTimeoutMs(),
                 clientRequest.createdTimeMs(),
                 clientRequest.destination(),
                 clientRequest.callback(),
                 clientRequest.expectResponse(),
                 isInternalRequest,
                 request,
                 send,
                 sendTimeMs);
        }

        public InFlightRequest(RequestHeader header,
                               int requestTimeoutMs,
                               long createdTimeMs,
                               String destination,
                               RequestCompletionHandler callback,
                               boolean expectResponse,
                               boolean isInternalRequest,
                               AbstractRequest request,
                               Send send,
                               long sendTimeMs) {
            this.header = header;
            this.requestTimeoutMs = requestTimeoutMs;
            this.createdTimeMs = createdTimeMs;
            this.destination = destination;
            this.callback = callback;
            this.expectResponse = expectResponse;
            this.isInternalRequest = isInternalRequest;
            this.request = request;
            this.send = send;
            this.sendTimeMs = sendTimeMs;
        }

        /**
         * 收到响应，回调的时候据响应内容生成ClientResponse
         * @param response
         * @param timeMs
         * @return
         */
        public ClientResponse completed(AbstractResponse response, long timeMs) {
            return new ClientResponse(header, callback, destination, createdTimeMs, timeMs,
                    false, null, null, response);
        }

        /**
         * 当连接突然断开，disconnected方法会生成ClientResponse。
         * @param timeMs
         * @param authenticationException
         * @return
         */
        public ClientResponse disconnected(long timeMs, AuthenticationException authenticationException) {
            return new ClientResponse(header, callback, destination, createdTimeMs, timeMs,
                    true, null, authenticationException, null);
        }

        @Override
        public String toString() {
            return "InFlightRequest(header=" + header +
                    ", destination=" + destination +
                    ", expectResponse=" + expectResponse +
                    ", createdTimeMs=" + createdTimeMs +
                    ", sendTimeMs=" + sendTimeMs +
                    ", isInternalRequest=" + isInternalRequest +
                    ", request=" + request +
                    ", callback=" + callback +
                    ", send=" + send + ")";
        }
    }

}
